package com.caiyi.financial.nirvana.sms.gateway;

import com.caiyi.financial.nirvana.sms.jedis.JedisKey;
import com.caiyi.financial.nirvana.sms.jedis.JedisService;
import com.caiyi.financial.nirvana.sms.util.ThreadPoolUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class GateWayThread<G extends DefaultGateWay> extends Thread {

    private static final Logger logger = LoggerFactory.getLogger(GateWayThread.class);

    // 是否存活
    private boolean alive = true;

    // 网关通道
    private G gateWay;

    // 网关ID
    private int gateWayId;

    GateWayThread(String name, G gateWay) {
        super(name);
        this.gateWay = gateWay;
        this.gateWayId = gateWay.info.getId();
        setDaemon(false);
    }

    /**
     * 安全退出线程
     */
    void kill() {
        this.gateWay.close();
        this.gateWay = null;
        this.alive = false;
    }

    @Override
    public void run() {
        String industry = String.format(JedisKey.SMS_QUEUE_INDUSTRY_MSG_SEND, gateWayId);
        String marketing = String.format(JedisKey.SMS_QUEUE_MARKETING_MSG_SEND, gateWayId);
        while (alive) {
            try {
                if (gateWay != null && !gateWay.login) {
                    gateWay.doConnect();
                    Thread.sleep(3000);
                }
                if (gateWay != null && gateWay.login) {
                    String message = JedisService.rpop(industry);
                    if (StringUtils.isNotBlank(message)) {
                        SendTask task = new SendTask(message, gateWay);
                        ThreadPoolUtil.EXECUTOR_INDUSTRY_MESSAGE.execute(task);
                        continue;
                    }
                    message = JedisService.rpop(marketing);
                    if (StringUtils.isNotBlank(message)) {
                        SendTask task = new SendTask(message, gateWay);
                        ThreadPoolUtil.EXECUTOR_MARKET_MESSAGE.execute(task);
                        continue;
                    }
                    Thread.sleep(1000);
                }
            } catch (Exception e) {
                try {
                    Thread.sleep(1000);
                } catch (Exception e1) {
                }
                logger.error("GateWayThread error gatewayName : {}" + gateWay.info.getName());
            }
        }
        logger.info("GateWayThread {} close.", Thread.currentThread().getName());
    }

    // 获取通道
    public G getGateWay() {
        return gateWay;
    }

    // 设置通道
    public void setGateWay(G gateWay) {
        this.gateWay = gateWay;
    }

}
